package com.heaven.service.impl;

import com.heaven.constant.MessageConstants;
import com.heaven.core.dao.MessageDao;
import com.heaven.core.entity.MessageEntity;
import com.heaven.core.entity.UserEntity;
import com.heaven.core.framework.PzServiceImpl;
import com.heaven.customerServer.customerController.vo.message.ChatHistoryRes;
import com.heaven.customerServer.customerController.vo.message.ChatTarget;
import com.heaven.exception.BizCodeEnum;
import com.heaven.exception.RRException;
import com.heaven.service.MessageService;
import com.heaven.service.UserService;
import com.heaven.utils.BeanUtils;
import com.heaven.utils.RedisUtils;
import jakarta.annotation.Resource;
import org.redisson.Redisson;
import org.redisson.api.RLock;
import org.springframework.data.domain.Sort;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;

@Service("messageService")
public class MessageServiceImpl extends PzServiceImpl<MessageDao, MessageEntity> implements MessageService {
    @Resource
    private RedisUtils redisUtils;
    @Resource
    private UserService userService;
    @Resource
    private Redisson redisson;

    @Override
    public List<ChatTarget> getChatTargets(String userId) {
        Set<String> targetIds = new HashSet<>(); // 存储聊天记录的目标用户id
        // 查缓存
        List<String> keys1 = redisUtils.keys(MessageConstants.MESSAGE_CACHE + "*:" + userId+"-*").stream().toList();
        keys1.forEach(key -> {
            String id = key.split(":")[2].split("-")[1];
            targetIds.add(id);
        });
        List<String> keys2 = redisUtils.keys(MessageConstants.MESSAGE_CACHE + "*:" + "*-"+userId).stream().toList();
        keys2.forEach(key -> {
            String id = key.split(":")[2].split("-")[0];
            targetIds.add(id);
        });

        // 查数据库
        UserEntity own = userService.selectById(userId);
        RLock rLock = redisson.getReadWriteLock(MessageConstants.MESSAGE_CHAT_TARGET_KEY + userId).writeLock();
        rLock.lock();
        Set<String> storeIds = own.getChatTargetIds();
        rLock.unlock();
        if(storeIds != null){
            targetIds.addAll(storeIds);
        }

        return targetIds.stream().map(id -> {
            UserEntity user = userService.selectById(id);
            user = userService.initUser(user);
            ChatTarget chatTarget = new ChatTarget();

            // 获取与id用户的消息统计情况
            Query query = Query.query(Criteria.where("from_id").is(id))
                    .with(Sort.by(Sort.Direction.DESC, "send_time"));
            final MessageEntity[] messageEntity = {selectOne(query)};  //记录其他人发来的消息以便统计未读消息数

            AtomicLong unReadCount = new AtomicLong(selectCount(Query.query
                    (Criteria.where("from_id").is(id).and("status").is(0))));
            // 缓存中存储的来自用户id的消息keys
            List<String> keys = redisUtils.keys(MessageConstants.MESSAGE_CACHE + "*:" + id + "-" + userId).stream().toList();
            keys.forEach(key -> {
                MessageEntity msg = redisUtils.getCacheObject(key);
                if (messageEntity[0] == null) messageEntity[0] = msg;
                else if (msg.getSendTime().after(messageEntity[0].getSendTime())) {
                    messageEntity[0] = msg;
                }
                if (msg.getStatus() == 0) {
                    unReadCount.getAndIncrement();
                }
            });

            if (messageEntity[0] != null) {
                BeanUtils.copyProperties(messageEntity[0], chatTarget);
                chatTarget.setLastMsg(messageEntity[0].getMsgContent());
            }

            chatTarget.setUnReadNum(unReadCount.intValue());

            chatTarget.setSubscribed(own.getSubscribeUsers().contains(id));

            // 获取id用户的基本信息
            ChatTarget.UserInfo userInfo = new ChatTarget.UserInfo();
            BeanUtils.copyProperties(user, userInfo);
            chatTarget.setUserInfo(userInfo);
            return chatTarget;
        }).sorted(Comparator.comparing(ChatTarget::getSendTime)) // 按照发送时间降序排序
                .toList();
    }

    @Override
    public ChatTarget getUserInfo(String ownId, String targetUserId) {
        UserEntity userEntity = userService.selectById(targetUserId);
        UserEntity own = userService.selectById(ownId);
        if(userEntity == null || own == null){
            throw new RRException(BizCodeEnum.USER_NOT_EXIST,targetUserId);
        }

        ChatTarget chatTarget = new ChatTarget();
        ChatTarget.UserInfo userInfo = new ChatTarget.UserInfo();
        userEntity = userService.initUser(userEntity);
        BeanUtils.copyProperties(userEntity, userInfo);
        chatTarget.setUserInfo(userInfo);
        chatTarget.setSubscribed(own.getSubscribeUsers().contains(targetUserId));
        AtomicLong unReadCount = new AtomicLong(selectCount(Query.query(Criteria.where("from_id").is(targetUserId).and("status").is(0))));

        AtomicReference<MessageEntity> messageEntity = new AtomicReference<>(selectOne(Query.query(Criteria.where("from_id").is(targetUserId))
                .with(Sort.by(Sort.Direction.DESC, "send_time")).limit(1)));
        List<String> keys = redisUtils.keys(MessageConstants.MESSAGE_CACHE + "*:" + targetUserId+"-"+ownId).stream().toList();
        keys.forEach(key -> {
            MessageEntity msg = redisUtils.getCacheObject(key);
            if (msg.getStatus() == 0) {
                unReadCount.getAndIncrement();
            }
            if(messageEntity.get() == null ){
                messageEntity.set(msg);
            } else if(msg.getSendTime().after(messageEntity.get().getSendTime())){
                messageEntity.set(msg);
            }
        });
        chatTarget.setUnReadNum(unReadCount.intValue());
        if(messageEntity.get() != null){
            BeanUtils.copyProperties(messageEntity.get(), chatTarget);
            chatTarget.setLastMsg(messageEntity.get().getMsgContent());
        }

        return chatTarget;
    }

    @Override
    public ChatHistoryRes getChatHistory(String userId, String targetUserId, int pageSize, int pageNum) {
        Query query = Query.query(new Criteria()
                .orOperator(Criteria.where("from_id").is(userId).and("to_id").is(targetUserId),
                        Criteria.where("from_id").is(targetUserId).and("to_id").is(userId)));

        // 计数总聊天记录数
        long total = selectCount(query);
        Set<String> cacheKeys = new HashSet<>(redisUtils.keys(MessageConstants.MESSAGE_CACHE + "*:" + userId + "-" + targetUserId));
        cacheKeys.addAll(redisUtils.keys(MessageConstants.MESSAGE_CACHE + "*:" + targetUserId+"-"+userId));

        total += cacheKeys.size();

        // 计算已经获取的记录数
        long alreadyCount = (long) pageSize * (pageNum - 1);
        if(alreadyCount >= total) {
            return new ChatHistoryRes(total, new ArrayList<>());
        } else {
            // 查聊天记录
            List<MessageEntity> chatMsgList = new ArrayList<>();
            query.with(Sort.by(Sort.Direction.DESC, "send_time"));

            if(cacheKeys.isEmpty()){
                //1、缓存中没有
                query.skip(alreadyCount).limit(pageSize);
                chatMsgList.addAll(selectList(query));
            }else{
                //2、缓存中有
                for (String key : cacheKeys) {
                    if(chatMsgList.size() >= pageSize) break;
                    MessageEntity msg = redisUtils.getCacheObject(key);
                    if(msg.getFromId().equals(targetUserId) && msg.getStatus() == 0) {
                        msg.setStatus(1);
                        save(msg);
                    }
                    chatMsgList.add(msg);
                    redisUtils.deleteObject(key);
                }

                if(cacheKeys.size() < pageSize ) {
                    //3、缓存中有，但数量不够（将缓存中的消息取出入库）
                    query.skip(alreadyCount + chatMsgList.size()).limit(pageSize - cacheKeys.size());
                    chatMsgList.addAll(selectList(query));
                }
            }

            // 将 chatMsgList 里的未读消息全更新为已读状态
            List<MessageEntity> list = chatMsgList.stream().peek(msg -> {
                if (msg.getStatus() == 0 && msg.getFromId().equals(targetUserId)) {
                    msg.setStatus(1);
                    updateOne(Query.query(Criteria
                            .where("_id").is(msg.getMessageId())), Update.update("status", 1));
                }
            }).sorted(Comparator.comparing(MessageEntity::getSendTime)).toList();

            return new ChatHistoryRes(total, list);
        }
    }

}
